package com.flink.datastreamapi.function;

import com.flink.datastreamapi.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//定义接口的实现类，传入如参数的方式
public class FunctionUDFDemo4 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<WaterSensor> stream = env.fromElements(

                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_1", 12L, 12),
                new WaterSensor("sensor_2", 2L, 2),
                new WaterSensor("sensor_3", 3L, 3)
        );

        DataStream<WaterSensor> filter = stream.filter(new CustomFilterImpl("sensor_2"));

        filter.print();
        env.execute();
    }

    public static class CustomFilterImpl implements FilterFunction<WaterSensor> {

        private String id;

        public CustomFilterImpl(String id){
            this.id=id;
        }

        @Override
        public boolean filter(WaterSensor e) throws Exception {
            return e.id.equals(this.id);
        }
    }
}
